[Platform] Introduce asynchronous jobs and move MiniMax polling out of the result converter - #2408
Merged
chr-hertel merged 1 commit intoSep 20, 2026
Conversation
wachterjohannes
requested review from
OskarStark and
chr-hertel
as code owners
August 15, 2026 20:06
wachterjohannes
force-pushed
the
feature/platform-async-jobs
branch
from
August 15, 2026 20:25
c7ba9f6 to
3fae510
Compare
wachterjohannes
commented
Aug 15, 2026
| * | ||
| * @internal | ||
| */ | ||
| final class TarArchive |
Member
Author
There was a problem hiding this comment.
the API delivery a tar file - to avoid the dependency to a tar lib i have implemented this little class
chr-hertel
reviewed
Aug 16, 2026
chr-hertel
reviewed
Aug 16, 2026
wachterjohannes
force-pushed
the
feature/platform-async-jobs
branch
from
August 24, 2026 21:21
2ad2088 to
7d3bcd1
Compare
wachterjohannes
force-pushed
the
feature/platform-async-jobs
branch
from
September 1, 2026 19:44
7d3bcd1 to
5bfd3b2
Compare
chr-hertel
requested changes
Sep 11, 2026
chr-hertel
left a comment
Member
There was a problem hiding this comment.
We have an isolation problem here that we should resolve still.
Symptoms:
$result instanceof JobResultinTypedResultTrait::as()$result instanceof JobResultinProvider::invoke()
I think this should be handled already down below in the ModelClient/ResultConverter layer while constructing the JobResult. We don't have the provider name there, but strictly speaking that is irrelevant. from a logic point of view that layer "knows" where it is and how to move on 🤔
wachterjohannes
force-pushed
the
feature/platform-async-jobs
branch
from
September 20, 2026 21:22
5bfd3b2 to
4b807a6
Compare
wachterjohannes
force-pushed
the
feature/platform-async-jobs
branch
from
September 20, 2026 21:31
4b807a6 to
9ac8e0e
Compare
…f the result converter
chr-hertel
force-pushed
the
feature/platform-async-jobs
branch
from
September 20, 2026 22:27
9ac8e0e to
3bf8906
Compare
Member
|
Thank you @wachterjohannes. |
This was referenced Sep 20, 2026
chr-hertel
added a commit
to chr-hertel/ai
that referenced
this pull request
Sep 21, 2026
chr-hertel
added a commit
that referenced
this pull request
Sep 21, 2026
…r-hertel) This PR was squashed before being merged into the main branch. Discussion ---------- [Platform] Fix asynchronous job follow-ups from #2408 | Q | A | ------------- | --- | Bug fix? | no | New feature? | no | Docs? | yes | Issues | - | License | MIT Follow-ups on #2408, found while reviewing it after the merge. Ships with #2408 in 0.14, so nothing released changes and no `UPGRADE.md` entry is added — the existing ones are corrected to describe what will actually ship. * MiniMax answers a healthy poll with `status_code: 0` **and** `status_msg: "success"`, which was read as a failure message, so every running job carried `"success"` in `JobStatus::getError()`. Verified against the live API. * A query MiniMax rejects with HTTP 200 carries the reason in `base_resp` and no state at all, which mapped to `UNKNOWN` (non-terminal) — the runner then polled it for the whole budget, up to ten minutes for a video handle. It now ends the job, while a state merely spelled in a way the bridge does not know stays non-terminal. * `JobRunner` turned the budget into a poll count, so the requests themselves were free: `maxDuration: 600` meant 600 requests plus 599 sleeps. It now waits against a clock deadline. * `JobClientInterface::supports()` had no caller anywhere; the runner asks before polling. * `MiniMaxResultConverter` held a `MiniMaxJobClient` only to read a name off it, building one with `new EventSourceHttpClient()` and an empty API key when given none. It takes the name; `MiniMaxJobClient::createHandle()` and its provider argument go. * Docs: the autowiring example used `$minimaxJobClient`, but the registered alias is `JobClientInterface $minimax`, so it never autowired — checked by compiling a container. Also a nullable `getProvider()` passed to `ContainerInterface::get()`, an unassigned `$id`, and examples using `createProvider()` where every other MiniMax example uses `createPlatform()`. * The 0.14 rebase had replaced the `CHANGELOG` heading underline with `==` in two files; `examples/minimax/.gitignore` missed the artifacts the new examples write. All nine MiniMax examples were run against the live API: async speech (9s, real mp3 out of the tar), video generation (83s, 792 KB mp4) and the resume flow across four processes from the stored handle. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Commits ------- e7af258 [Platform] Fix asynchronous job follow-ups from #2408
chr-hertel
added a commit
to chr-hertel/ai
that referenced
this pull request
Sep 21, 2026
An alternative to symfony#1802: batch is not its own stack of BatchJob/BatchManager/ BatchStatus classes, but the asynchronous job that symfony#2408 introduced - a batch is a job that happens to carry many requests. Submitting goes through `Platform::invoke()` with several inputs instead of one, keyed by the identifier each result is reported back under, and returns a plain `JobHandle`: $handle = $platform->invoke('gpt-4o-mini', [ 'capital-fr' => new MessageBag(Message::ofUser('What is the capital of France?')), 'capital-de' => new MessageBag(Message::ofUser('What is the capital of Germany?')), ], ['batch' => true])->asJob(); Resolving it is `JobClientInterface`, so storing the handle, polling from a worker, `JobRunner`, the exceptions and the bundle wiring all come for free. A finished batch converts into a `BatchResult` of `BatchItem`s, read from the download as they are iterated. A successful item carries the ordinary result the request would have produced synchronously, built by the bridge's own converter, so tool calls and structured output survive the detour.
chr-hertel
added a commit
to chr-hertel/ai
that referenced
this pull request
Sep 22, 2026
An alternative to symfony#1802: batch is not its own stack of BatchJob/BatchManager/ BatchStatus classes, but the asynchronous job that symfony#2408 introduced - a batch is a job that happens to carry many requests. Submitting goes through `Platform::invoke()` with several inputs instead of one, keyed by the identifier each result is reported back under, and returns a plain `JobHandle`: $handle = $platform->invoke('gpt-4o-mini', [ 'capital-fr' => new MessageBag(Message::ofUser('What is the capital of France?')), 'capital-de' => new MessageBag(Message::ofUser('What is the capital of Germany?')), ], ['batch' => true])->asJob(); Resolving it is `JobClientInterface`, so storing the handle, polling from a worker, `JobRunner`, the exceptions and the bundle wiring all come for free. A finished batch converts into a `BatchResult` of `BatchItem`s, read from the download as they are iterated. A successful item carries the ordinary result the request would have produced synchronously, built by the bridge's own converter, so tool calls and structured output survive the detour.
chr-hertel
added a commit
that referenced
this pull request
Sep 22, 2026
… (chr-hertel) This PR was merged into the main branch. Discussion ---------- [Platform] Convert polling bridges to asynchronous jobs | Q | A | ------------- | --- | Bug fix? | no | New feature? | yes | Docs? | yes | Issues | - | License | MIT Follow-up to #2408: the Higgsfield, Venice and Replicate bridges solved async work with their own polling loops. They now return a `JobResult` carrying a `JobHandle`, resolved through a per-bridge `JobClientInterface` — same shape as MiniMax. - `HiggsfieldJobClient`, `VeniceJobClient`, `ReplicateJobClient` + `Factory::createJobClient()` - ai-bundle registers `ai.platform.job_client.higgsfield` / `.venice` (tagged + autowired) - Removed the now-dead `clock` / `pollingInterval` / `max_polling_attempts` / `polling_interval_seconds` args While at it, `JobHandle` gained `getPollInterval()` next to `getMaxDuration()` — how often a job is worth asking about is the bridge's knowledge just like how long it runs — and `JobRunner` gained a `maxPolls` ceiling, which is the caller's (a rate limit, an API bill). Both resolve `call > runner > handle > default`. **Needs the `BC Break` label** — Replicate is released, and its invocations no longer return a `TextResult` directly (`UPGRADE.md` entry included). Higgsfield and Venice were both added in the unreleased 0.14, so they get no upgrade entry. Cassettes were extended by hand rather than re-recorded (no API keys): `getStatus()` + `getResult()` means one extra poll of the finished job, so each gained a duplicate of its final terminal-status interaction — the same shape MiniMax's committed cassette already has. Also passes the examples' clock to the MiniMax job runners, which were sleeping for real on replay: `minimax/text-to-video.php` went from 52s to 0.06s, and the whole replay suite from 67s to 16s. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Commits ------- af06ee0 [Platform] Convert polling bridges to asynchronous jobs
chr-hertel
added a commit
to chr-hertel/ai
that referenced
this pull request
Sep 22, 2026
An alternative to symfony#1802: batch is not its own stack of BatchJob/BatchManager/ BatchStatus classes, but the asynchronous job that symfony#2408 introduced - a batch is a job that happens to carry many requests. Submitting goes through `Platform::invoke()` with several inputs instead of one, keyed by the identifier each result is reported back under, and returns a plain `JobHandle`: $handle = $platform->invoke('gpt-4o-mini', [ 'capital-fr' => new MessageBag(Message::ofUser('What is the capital of France?')), 'capital-de' => new MessageBag(Message::ofUser('What is the capital of Germany?')), ], ['batch' => true])->asJob(); Resolving it is `JobClientInterface`, so storing the handle, polling from a worker, `JobRunner`, the exceptions and the bundle wiring all come for free. A finished batch converts into a `BatchResult` of `BatchItem`s, read from the download as they are iterated. A successful item carries the ordinary result the request would have produced synchronously, built by the bridge's own converter, so tool calls and structured output survive the detour.
chr-hertel
added a commit
that referenced
this pull request
Sep 22, 2026
…nous jobs (chr-hertel) This PR was squashed before being merged into the main branch. Discussion ---------- [Platform] Add OpenAI batch requests on top of asynchronous jobs | Q | A | ------------- | --- | Bug fix? | no | New feature? | yes | Docs? | yes | Issues | Fix #1776 | License | MIT An alternative to #1802: now that #2408 landed asynchronous jobs, a batch is just a job that carries many requests — no `BatchJob`/`BatchManager`/`BatchStatus` stack needed. ```php $handle = $platform->invoke('gpt-4o-mini', [ 'capital-fr' => new MessageBag(Message::ofUser('What is the capital of France?')), 'capital-de' => new MessageBag(Message::ofUser('What is the capital of Germany?')), ], ['batch' => true])->asJob(); // later, in a worker, from the stored handle $jobClient = OpenAiFactory::createJobClient($apiKey); foreach ($jobClient->getResult($handle)->getContent() as $item) { echo $item->getId().': '.($item->isSuccess() ? $item->getResult()->getContent() : $item->getError()); } ``` * Storing the handle, polling, `JobRunner`, the exceptions and the bundle wiring all come from #2408 unchanged * `['batch' => true]` is handled by `Gpt\ModelClient` and delegated to `Batch\BatchClient`, like `MiniMaxClient` routes on `async` — no routing in `Provider` * A finished batch converts into a `Result\BatchResult` of `Result\BatchItem`s, read from the download as they are iterated; a successful item carries the ordinary result the request would have produced synchronously, built by the bridge's own converter (tool calls, structured output, finish reasons included) * An item without a result states *why* through a `Result\BatchItemCase` — `ERRORED` has to be fixed, `CANCELED`/`EXPIRED` were never sent and can simply be resubmitted — with the provider's own wording kept on `getRaw()`, the same split as `FinishReason` and `JobStatus` * A cancelled or expired batch still hands out what it finished, since fetching keys on the result file rather than on `completed` * `Batch\JobClient` adds `getRequestCounts()` and `cancel()` — on the bridge, not on `JobClientInterface` ### Does the result shape generalize? Checked the two obvious next bridges before settling it, since `BatchResult`/`BatchItem` sit in the Platform: * **Mistral** — the same envelope to the letter. Its SDK parses `custom_id` / `response.body` / `error` with `status_code >= 400` as the failure test, which is `Batch\JobClient::toItem()` line for line, and documents that envelope as invariant across endpoints. Statuses map onto `JobStateCase` without a gap. * **Anthropic** — submits inline rather than as a file (submit side only, per bridge) and reports four per-request outcomes: `succeeded`, `errored`, `canceled`, `expired`. That is what `BatchItemCase` is for. Its batch-level status is only `in_progress`/`canceling`/`ended`, and a cancelled batch also ends as `ended` with partial results — so keying result-fetching on the file rather than on a success status is not just convenient here, it is the only thing that works there. Request counts differ per provider (`{total, completed, failed}` vs `{processing, succeeded, errored, canceled, expired}` vs `{total, succeeded, failed}`), which is why `getRequestCounts()` stays on the bridge. ### Open point No `Capability::BATCH`. OpenAI gates batching per endpoint, not per model, and the option is handled by the endpoint's own model client, so the gate is structural and there is no list to keep up to date — an ineligible model is rejected by the API. Happy to add the capability instead if you prefer the pre-flight check. Scoped out for follow-ups: the Anthropic and Mistral bridges, and batched embeddings (same `BatchClient`, different endpoint). Commits ------- 5e61785 [Platform] Add OpenAI batch requests on top of asynchronous jobs
marco-jouwweb
pushed a commit
to marco-jouwweb/symfony-ai
that referenced
this pull request
Sep 23, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Asynchronous work already exists in this component — hand-rolled, and in the wrong layer:
Bridge\MiniMax\MiniMaxResultConverter::handleAsyncTask()polls a task withclock->sleep()up to 600 times, holding anHttpClientInterface, an API key and aClockInterfaceto do so — inside a class whose contract is a pureRawResult → Resultmapping.Bridge\Replicate\Client::request()does the same in the model client.Capability::TEXT_TO_SPEECH_ASYNCis already in the enum, used by exactly one bridge.So the question is not whether the component should support async work, but that it already does, twice, inconsistently. Today one
invoke()call blocks for minutes, the caller cannot observe the job or stop waiting, the operation does not survive the process that started it, and the state machine is not testable in isolation.This PR introduces the job as a platform primitive and moves MiniMax onto it. Replicate and batch endpoints are deliberately left for follow-ups.
Why "job" and not "batch"
Video generation, asynchronous speech, OpenAI/Anthropic message batches, Gemini long-running operations and Replicate predictions all share one mechanic: submit → identifier → poll → fetch. And the body you eventually get back is the same one a synchronous call would have returned, which is why
ResultConverterInterfaceis untouched here — an asynchronous result is not a new result type, only a different delivery.Batch is then "a job with N items" rather than a second abstraction:
JobHandle::$datacarries the per-item keys when that lands.What it looks like
The handle holds no client and no connection, only what is needed to ask the provider about the job again — so it can be stored and picked up elsewhere:
To simply block, hand it to a runner — which owns the one polling loop left in the component and takes its budget from the caller, seconds for speech and minutes for video, instead of a constant inside a bridge:
Pieces
Job\JobHandledata. No client, no closureJob\JobStateCase/Job\JobStatusFinishReason/FinishReasonCase. An unknown state maps toUNKNOWNand counts as non-terminal, so a state a provider adds later does not abort a running jobJob\JobClientInterfacegetStatus()/getResult(), exactly one request per call, never sleepsJob\JobRunnerClockInterface, poll interval and budgetResult\JobResultResultInterfacewhose content is the handle;DeferredResult::asJob()reads itJob\JobProviderInterfaceProviderInterfacestays untouchedProvidertakes an optionalJobClientInterfaceas its last constructor argument, and stamps the provider name onto the handle after conversion — a converter cannot know the name its provider was registered under, since that is aProviderargument andFactory::createProvider()lets it be overridden.Exception\JobFailedExceptioncarries the status;Exception\JobTimeoutExceptioncarries the handle, so a job that outlived its budget can be handed on rather than lost.Verified against the live API
Not only against mocks — every path was run against api.minimax.io:
ISO Media, MP4 Base Media v1Processingin a second,Success+ download in a third — resolved purely from the serialized handleRunning it also turned up two things mocks could not have shown, both fixed here:
MiniMax reports rejections with HTTP 200. The reason sits in
base_respand the payload keys are simply absent — an unknown voice ont2a_v2gives2054 "voice id not exist", an unsupported model onimage_generationgives2013 "invalid params, ...", an empty account gives1008 "insufficient balance", all with HTTP 200, while every success carriesstatus_code: 0. Previously this surfaced as an error about a missing array key — or, for an asynchronous task, as polling a task that never existed (task_id: 0is handed out as a placeholder) until the budget ran out. The converter now checksbase_respfor every endpoint. Chat is the exception: a bad model there does return a real HTTP 400, already covered byHttpStatusErrorHandlingTrait.The asynchronous speech endpoint does not return audio. It returns a tar bundling the mp3 with a
.titlesand an.extrafile, which the bridge previously labelledaudio/mpeg— soasFile('out.mp3')wrote an archive. The job client now unpacks the audio, making async and sync speech produce the same thing.TarArchiveis a small ustar reader: the archive uses the header'sprefixfield because the paths exceed the 100-byte name field, and the mp3 is not the first member, so both had to be handled.PharDatawould have meant a temp file plusext-pharas a new bridge dependency.The test fixture is a real archive produced by
tar --format ustarrather than hand-written, so the reader is not being tested against its own writer.Not in this PR
Client.php:60) — same shape, follow-up now that the primitive exists.custom_idcorrelation).Capability::ASYNC_JOB— theJobResulttype is already the signal, a capability would be a second truth.ai-bundlevia Messenger/Scheduler/symfony/webhook; Platform only provides submit/status/fetch.base_respcode throws aRuntimeExceptioncarrying the code and the provider's message. Mapping1004toAuthenticationExceptionand1002toRateLimitExceededExceptionwould be plausible, but only1008,2013and2054were actually observed and a half-map is worse than none. Happy to add it if wanted.BC break
Video generation and asynchronous speech return a
JobResultinstead of binary data, so reading them throughasBinary()/asFile()throws. Waiting became explicit.MiniMaxResultConverteralso lost its HTTP client, API key, endpoint and clock arguments. Both are documented inUPGRADE.md; code building the bridge throughBridge\MiniMax\Factoryis unaffected.